package jmind.core.rabbitmq;

import com.rabbitmq.client.AMQP;
import jmind.base.lang.IProperties;
import jmind.base.lang.shard.ConsistentHashLoadBalance;
import jmind.base.lang.shard.LoadBalance;
import jmind.base.lang.shard.RoundRobinLoadBalance;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeoutException;

/**
 * 集群模式的生产队列
 * Created by xieweibo on 2016/11/3.
 */
public class LoadBalanceProducer implements Producer {

    private final LoadBalance<RabbitmqProducer> loadBalance ;

    public LoadBalanceProducer(String names, IProperties p) throws IOException, TimeoutException {
        this(LoadBalance.Balance.RoundRobin,names,p,null,null,null,false);
    }
    public LoadBalanceProducer(String names, IProperties p, String queueName, String exchangeName, ExchangeType type, boolean durable) throws IOException, TimeoutException {
        this(LoadBalance.Balance.RoundRobin,names,p,queueName,exchangeName,type,durable);
    }
    public LoadBalanceProducer(LoadBalance.Balance balance, String names, IProperties p, String queueName, String exchangeName, ExchangeType type, boolean durable) throws IOException, TimeoutException {
        final List<RabbitmqProducer> producers = new CopyOnWriteArrayList<RabbitmqProducer>();
        for(String name:names.split(",")){
            producers.add(new RabbitmqProducer(name,p,queueName,exchangeName,type,durable));
        }
        if(balance==LoadBalance.Balance.RoundRobin){
            loadBalance=new RoundRobinLoadBalance<RabbitmqProducer>(producers);
        }else{
            loadBalance=new ConsistentHashLoadBalance<RabbitmqProducer>(producers);
        }
    }

    @Override
    public boolean produce(String routingKey, String message) {
        return loadBalance.getShard(routingKey).produce(routingKey,message);
    }

    @Override
    public boolean produce(String exchange, String routingKey, String message) {
        return loadBalance.getShard(routingKey).produce(exchange,routingKey,message);
    }

    @Override
    public boolean produce(String exchange, String routingKey, int delayTime, String message) {
        return loadBalance.getShard(routingKey).produce(exchange,routingKey,delayTime,message);
    }

    @Override
    public boolean produce(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body) {
        return loadBalance.getShard(routingKey).produce(exchange,routingKey,props,body);
    }
}
